package worker;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.PrintStream;
import java.lang.management.ManagementFactory;
import java.lang.management.ThreadMXBean;
import java.math.BigInteger;
import java.net.InetAddress;
import java.net.Socket;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.Semaphore;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.swing.text.html.MinimalHTMLWriter;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import comm.ConnectionType;
import comm.InfrastructureInformation;
import comm.RunMode;
import comm_2.Comm;
import comm_2.Streams;
import comm_2.SystemInformation;

import result.Result;
import screensaver.ScreenSaverWindow;
import tasks.CachedTask;
import tasks.Task;
import tasks.TaskId;
import worker.Worker.WorkerThread.TaskThread;

public class Worker extends Thread {

	String masterAddress;
	PrintStream out2;
	boolean keepRunning = true;
	boolean screenSaverMode = false;
	private float workerEvaluation = 0;;
	private int mainWorkerID;
	private InfrastructureInformation infrastructureInformation;

	private Semaphore numberOfAllowedThreads;
	private Semaphore numberOfAllowedCache;
	private Semaphore oneTaskAtTime;

	private static Logger logToMasterServer = Logger.getLogger(Worker.class
			.getName());
	private static Logger logToClient = Logger
	.getLogger(Worker.class.getName());

	private ClassLoaderObjectInputStream socketIn;
	private ObjectOutputStream socketOut;

	private ObjectInputStream getCodeInputStream;
	private ObjectOutputStream getCodeOutputStream;
	private SystemInformation localhostinfo;
	private WorkerData workerData;
	private int numberofCores;
	private int cacheSize;
	private int numberOfWorkerThreads;
	private HashMap<TaskId, TaskThread> currentTask = new HashMap<TaskId, TaskThread>();
	private LinkedList<CachedTask> cachedTaskList = new LinkedList<CachedTask>();
	private static ThreadMXBean threadBean = ManagementFactory
	.getThreadMXBean();
	private final static int retryConnect = 5000;
	private RunMode runMode = RunMode.minimumWorkerLoad;
	ExecutorService execSvc;

	public void startWorker(InfrastructureInformation infrastructureInformation,
			PrintStream out, int mainWorkerID,
			int numberOfCores) throws InterruptedException, IOException {

		this.out2 = out;
		this.mainWorkerID = mainWorkerID;
		this.numberofCores = numberOfCores;
		this.infrastructureInformation = infrastructureInformation;
		
		this.workerData = new WorkerData();
		System.out.println(threadBean.getCurrentThreadCpuTime());
		System.out.println(threadBean.getCurrentThreadUserTime());
		System.out.println(threadBean.getDaemonThreadCount());

		this.runMode = infrastructureInformation.getRunMode();
		this.cacheSize = this.numberofCores + (numberOfCores / 2);
		System.out.println("CACHE" + cacheSize + " " + numberOfCores);
		this.numberOfAllowedCache = new Semaphore(cacheSize); // cachesize
		this.numberOfAllowedThreads = new Semaphore(numberOfCores);
		this.oneTaskAtTime = new Semaphore(1);
		this.execSvc = Executors.newFixedThreadPool( numberofCores );


		getLocalHostInfo();
		connectCodeServer();
		connectMasterServer();

this.start();
	}

	public Worker() {
	}

	public void run() {

		try {
			serve();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (ClassNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		// benchmark();

		//	logToMasterServer.log(Level.SEVERE, e.toString());


		// TODO Auto-generated catch block
		//	logToMasterServer.log(Level.SEVERE, e.toString());



	}
	public WorkerData getWorkerData(){
		return this.workerData;
	}

	private void connectCodeServer() throws InterruptedException, IOException {
		Streams InOut = null;
		System.out.println("Trying to connect to CodeServer: "
				+ infrastructureInformation.getCodeServerIPaddress()
				+ " on port "
				+ infrastructureInformation.getCodeServerEncryptedPort());
		while (InOut == null){


			InOut = new Comm(
					infrastructureInformation.getCodeServerEncryptedPort(),
					infrastructureInformation.getCodeServerIPaddress())
			.startConnectionToServer();

			if (InOut != null) {
				this.getCodeInputStream = InOut.returnObjectInputStream();
				this.getCodeOutputStream = InOut.returnObjectOutputStream();

				this.getCodeOutputStream
				.writeObject(ConnectionType.CLASS_REQUESTER);
				return;
			} else {
				System.out.println("Comm has problems!!");
			}


			Thread.sleep(retryConnect);
		}

	}

	private void connectMasterServer() throws InterruptedException, IOException {
		Streams InOut = null;
		while(InOut==null){


			InOut = new Comm(
					infrastructureInformation.getMasterServerEncryptedPort(),
					infrastructureInformation.getMasterServerIPaddress())
			.startConnectionToServerObject(this.getCodeInputStream,
					this.getCodeOutputStream);

			if (InOut != null) {
				this.socketIn = InOut.returnClassLoaderObjectInputStream();
				this.socketOut = InOut.returnObjectOutputStream();

				this.socketOut.writeObject(ConnectionType.WORKER);
				return;
			} else {
				System.out.println("Comm has problems!!");
			}




			Thread.sleep(retryConnect);
		}



	}

	private void benchmark() {

		int numberOfProcessors = this.workerData.getNumberOfProcessors();
		calculatorFIB[] c = new calculatorFIB[16];

		for (int i = 0; i < numberOfProcessors; i++) {
			c[i] = new calculatorFIB(160000);
			c[i].start();

		}

	}

	private class calculatorFIB extends Thread {

		int nr;

		public calculatorFIB(int nr) {
			this.nr = nr;

		}

		public void run() {
			BigInteger sub2 = new BigInteger("1");
			BigInteger sub1 = new BigInteger("0");

			BigInteger total = new BigInteger("0");

			int p = 0;
			long start = System.currentTimeMillis();
			while (p < nr) {

				total = sub1;
				sub1 = sub1.add(sub2);
				sub2 = total;
				p++;
			}
			long end = System.currentTimeMillis();
			System.out.println("Terminou:" + this.getId() + " Tempo: "
					+ (end - start) / 1000 + " - " + sub2);
			setCpuEvaluation((end - start) / 1000);
		}

	}

	private void setCpuEvaluation(float eval) {

		this.workerEvaluation = eval / workerData.getNumberOfProcessors();
		System.out.println("evaluation" + workerEvaluation);

	}

	private void getLocalHostInfo() {
		Runtime runtime = Runtime.getRuntime();
		int numberOfProcessors;
		String operatingSystem;
		operatingSystem = System.getProperty("os.name");

		numberOfProcessors = runtime.availableProcessors();
		try {
			this.workerData.setMainWorkerID(mainWorkerID);
			localhostinfo = new SystemInformation(numberOfProcessors,
					operatingSystem, InetAddress.getLocalHost());
			System.out.println(localhostinfo.toString());
			this.workerData.setOperatingSystem(operatingSystem);
			this.workerData.setWorkerAddress(InetAddress.getLocalHost()
					.toString());
			this.workerData.setNumberOfProcessors(numberOfProcessors);
			DateFormat dateFormat = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
			Date date = new Date();
			this.workerData.setStartTime(dateFormat.format(date));

			System.out.println(localhostinfo.toString());
		} catch (UnknownHostException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

	}

	private synchronized void addWork() {
		System.out.println("HERE");

		while (numberOfWorkerThreads >= numberofCores) {
			try {
				wait();
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}

			numberOfWorkerThreads++;
		}
		System.out.println("Numero de workers:" + numberOfWorkerThreads);
	}

	private synchronized void freeTask() {
		numberOfAllowedThreads.release();

	}

	private void addTaskcacheList(TaskId taskid, Task task) {
		System.out.println("add");
		synchronized (this) {
			cachedTaskList.add(new CachedTask(taskid, task));
			//System.out.println(taskid.getTaskId());
			notify();
		}

	}

	private CachedTask getTask() {
		CachedTask cachedTask = null;

		synchronized (this) {
			while (cachedTaskList.size() == 0)
				try {

					wait();
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}

				cachedTask = cachedTaskList.poll();
		}
		numberOfAllowedCache.release();
		return cachedTask;
	}

	private void serve() throws IOException, ClassNotFoundException {
		keepRunning = true;
		socketOut.reset();
		socketOut.writeObject(workerData);
		workerData = (WorkerData) socketIn.readObject();
		new getTasks().start();
		new FeedWorker().start();

		ConnectionType type;
		while (true) {
			type = (ConnectionType) socketIn.readObject();
			System.out.println(type);
			switch (type) {
			case FEED_WORKER:

				TaskId tid = (TaskId) socketIn.readObject();
				socketIn.setProblemAndVersion(tid.getProblemNumber(),
						tid.getVersion());
				Task task = (Task) socketIn.readObject();
				addTaskcacheList(tid, task);

				oneTaskAtTime.release();
				break;
			case CLIENT_CANCEL_ALL_TASKS:
				long taskID = (Long) socketIn.readObject();
				System.out.println("Tenta matar:" + taskID);
				synchronized (currentTask) {

					Set<Entry<TaskId, TaskThread>> set = currentTask.entrySet();
					Iterator<Entry<TaskId, TaskThread>> i = set.iterator();
					while (i.hasNext()) {
						Map.Entry me = (Map.Entry) i.next();
						TaskId tidlocal = (TaskId) me.getKey();
						System.out.println(tidlocal.getTaskId());
						if (tidlocal.getTaskId() == taskID) {
							Thread th = (Thread) me.getValue();
							System.out.println(th.getState());

							th.interrupt();
							th.stop();
							i.remove();
							System.out.println(th.getState());


							freeTask();
							System.out.println("Matou Thread" + th.getId());
							synchronized (socketOut) {
								socketOut
								.writeObject(ConnectionType.WORKER_CANCELED_TASK_OK);
								socketOut.writeObject(taskID);
								System.out.println("KILLED:" + taskID);
							}

						}
					}

				}
				// System.out.println("saiu");
				break;
			case WORKER_CANCEL_CLIENT_TASK:
				break;
			case PING:
				synchronized (socketOut) {
					socketOut.writeObject(ConnectionType.PONG);

				}
			}

		}
	}

	class getTasks extends Thread {

		public void run() {
			while (true) {
				try {
					numberOfAllowedCache.acquire();
					oneTaskAtTime.acquire();
					// System.out.println("ASk for task"+cacheCounter++);
					synchronized (socketOut) {
						socketOut.writeObject(ConnectionType.WORKER_FEED);
					}

				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}

			}
		}
	}

	class FeedWorker extends Thread {
		private Task task;
		private TaskId tid;
		int nr = 0;

		public FeedWorker() {

		}

		public void run() {
			while (true) {
				try {
					numberOfAllowedThreads.acquire();
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				CachedTask cachedTask = getTask();
				// System.out.println("I:" + nr++);
				execSvc.execute( new WorkerThread(cachedTask.getTask(),cachedTask.getTaskid()));
				//wt.start();
			}
		}
	}

	class IdleController extends Thread {
		private Task task;
		private TaskId tid;
		int nr = 0;

		public IdleController() {

		}

		public void run() {
			while (true) {
				// detect iddle
				// idle out -> terminate tasks
				//
				// System.out.println("I:" + nr++);
				try {

					// System.out.println("aki");
					synchronized (socketOut) {

						socketOut.writeObject(ConnectionType.WORKER_FEED);

					}
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}

			}
		}
	}

	class WorkerThread implements Runnable
	{

		private Task task;
		private TaskId tid;

		public WorkerThread(Task task, TaskId tid) {
			this.task = task;
			this.tid = tid;
		}

		public void run() {
			boolean flag = false;
			long elapsedTime = 0L;
			DateFormat dateFormat = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
			TaskThread thread = new TaskThread(task, tid);
			synchronized (currentTask) {
				currentTask.put(tid, thread);
			}
			synchronized (workerData) {
				workerData.setRunning(true);
				workerData.setWorkerStatus("Starting...");

			}
			try {

				while (!Thread.currentThread().isInterrupted() && !flag) {

					long id = Thread.currentThread().getId();
					long startTime = System.currentTimeMillis();
					thread.start();

					try {
						thread.join();
					} catch (InterruptedException e) {
						// TODO Auto-generated catch block

						e.printStackTrace();
					}

					flag = true;
					elapsedTime = System.currentTimeMillis() - startTime;
					// System.out.println("Elapsed time - task related:"+elapsedTime);

					Date date = new Date();
					String endTime = dateFormat.format(date);
					synchronized (workerData) {
						workerData.setNumberOfTasksProcessed();
						workerData.setTotalTimeSpentFarming(elapsedTime);
						workerData.setRunning(false);
						workerData.setLastTaskTime(elapsedTime);
						workerData.setEndTime(endTime);
						workerData.setWorkerStatus("Stoped...");
						if (task.getResult() != null) {
							synchronized (socketOut) {
								socketOut.writeObject(ConnectionType.WORKER_RESULTS);
								socketOut.writeObject(tid);
								socketOut.writeObject(task.getResult());
								socketOut.writeObject(workerData);
								socketOut.reset();
							}
						}
					}
					synchronized (currentTask) {
						currentTask.remove(tid);
						freeTask();
					}

				}
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			long end = System.currentTimeMillis();
			System.out.println("___________DONE__________ " + tid);
		}

		class TaskThread extends Thread {
			private Task task;
			private TaskId tid;

			public TaskThread(Task task, TaskId tid) {
				this.task = task;
				this.tid = tid;
			}

			public void run() {
				try {
					long startCpuTime = threadBean.getCurrentThreadCpuTime();
					task.run();
					long cpuTime = (threadBean.getCurrentThreadCpuTime() - startCpuTime) / 1000000L;
					// System.out.println("Task number "+tid.getTaskId()+" CPU Time (ms):"+cpuTime);
				} catch (Throwable e) {
					System.out.println(e);
					this.interrupt();
				}

			}

		}
	}

}
